/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.commons.speeder.monitor;

import cn.ac.iie.di.commons.speeder.lifecycle.LifeCycle;
import cn.ac.iie.di.commons.speeder.lifecycle.LifeCycleException;
import cn.ac.iie.di.commons.speeder.monitor.handler.StatHandler;
import cn.ac.iie.di.commons.speeder.monitor.handler.impl.EmptyStatHandler;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 *
 * @author zwlin
 */
public class SpeedMonitor implements LifeCycle {

    private boolean activated;
    private Timer timer;
    private List<StatHandler<Long>> handlers;
    private StatHandler<String> errHandler;
    private int intervalSec;
    private int delaySec;

    private ConcurrentHashMap<String, AtomicLong> cache;

    public SpeedMonitor(int intervalSec, int delaySec) {
        this(intervalSec, delaySec, new ArrayList<>());
    }

    public SpeedMonitor(int intervalSec, int delaySec, StatHandler<Long> handler) {
        this(intervalSec, delaySec, Arrays.asList(handler));
    }

    public SpeedMonitor(int intervalSec, int delaySec, List<StatHandler<Long>> handlers) {
        this(intervalSec, delaySec, handlers, EmptyStatHandler.getInstance());
    }

    public SpeedMonitor(int intervalSec, int delaySec, List<StatHandler<Long>> handlers, StatHandler<String> errHandler) {
        if (delaySec < 0) {
            throw new LifeCycleException("delaySec < 0");
        }

        this.intervalSec = intervalSec;
        this.delaySec = delaySec;
        this.handlers = handlers;
        this.errHandler = errHandler;

        this.activated = false;
        this.timer = new Timer();
    }

    public void inc(String key, long size) {
        if (getIntervalSec() <= 0) {
            handleImediately(key, size);
        }
        checkLifeCycle();
        cache.compute(key, (k, v) -> Optional.ofNullable(v)
                .map(a -> a.addAndGet(size))
                .map(a -> v)
                .orElse(new AtomicLong(size)));
    }

    public void handleImediately(String key, long size) {
        checkLifeCycle();
        handlers.forEach(h -> h.accept(key, size));
    }

    @Override
    public void start() {
        if (!activated) {
            synchronized (this) {
                if (!activated) {
                    if (getIntervalSec() > 0) {
                        this.timer.scheduleAtFixedRate(new TimerTask() {
                            @Override
                            public void run() {
                                dump();
                            }
                        }, getDelaySec() * 1000, getIntervalSec() * 1000);
                        this.cache = new ConcurrentHashMap<>();
                    }
                    activated = true;
                }
            }
        }
    }

    @Override
    public void stop() {
        if (activated) {
            synchronized (this) {
                if (activated) {
                    activated = false;
                    this.timer.cancel();
                    dump();
                }
            }
        }
    }

    public void addHandler(StatHandler<Long> handler) {
        this.handlers.add(handler);
    }

    public void setErrorHandler(StatHandler<String> handler) {
        this.errHandler = handler;
    }

    public int getIntervalSec() {
        return intervalSec;
    }

    public void setIntervalSec(int intervalSec) {
        this.intervalSec = intervalSec;
    }

    public int getDelaySec() {
        return delaySec;
    }

    public void setDelaySec(int delaySec) {
        this.delaySec = delaySec;
    }

    private void dump() {
        ConcurrentHashMap<String, AtomicLong> last = cache;
        cache = new ConcurrentHashMap<>();
        try {
            Map<String, Long> map;
            if (last.isEmpty()) {
                map = Collections.EMPTY_MAP;
            } else {
                map = last.entrySet().stream().collect(
                        Collectors.toMap(e -> e.getKey(), e -> e.getValue().get()));
            }
            handlers.forEach(h -> h.accept(map));
        } catch (Exception | Error e) {
            try (StringWriter sw = new StringWriter(); PrintWriter pw = new PrintWriter(sw)) {
                e.printStackTrace(pw);
                errHandler.accept("error in monitor trace", sw.toString());
            } catch (Exception ee) {
                errHandler.accept("error in error trace", ee.getMessage() + "\n"
                        + Arrays.asList(ee.getStackTrace()).stream()
                                .map(s -> s.toString() + "\n")
                                .reduce((s, r) -> s + r).orElse(""));
            }
        }
    }

    private void checkLifeCycle() {
        if (!activated) {
            throw new LifeCycleException("not started.");
        }
    }
}
